package com.atguigu.kafka.flink;

import com.atguigu.kafka.flink.model.AccRes;
import com.atguigu.kafka.flink.model.Person;
import org.apache.flink.api.common.functions.AggregateFunction;

/**
 * 计算窗口内，每个人的月收入的平均值，以及月收入之差
 */
public class PersonAggregateFunction implements AggregateFunction<Person,PersonAggregateFunction.MyAcc, AccRes> {
    @Override
    public MyAcc createAccumulator() {
        return new MyAcc();
    }

    @Override
    public MyAcc add(Person person, MyAcc acc) {
        double avg = (acc.avgMoney * acc.count + person.getMoney()) / (1 + acc.count);
        if(acc.maxMoneyPerson==null||person.getMoney()>acc.maxMoneyPerson.getMoney()){
            acc.maxMoneyPerson=person;
        }
        if(acc.minMoneyPerson==null||person.getMoney()<acc.minMoneyPerson.getMoney()){
            acc.minMoneyPerson=person;
        }
        acc.count+=1;
        acc.avgMoney=avg;
        return acc;
    }

    @Override
    public AccRes getResult(MyAcc acc) {
        AccRes accRes=new AccRes();
        accRes.setAvg(acc.avgMoney);
        accRes.setCount(acc.count);
        accRes.setMaxDiffMoney(acc.maxMoneyPerson.getMoney()-acc.minMoneyPerson.getMoney());
        accRes.setMaxMoneyPersonName(acc.maxMoneyPerson.getName());
        accRes.setMinMoneyPersonName(acc.minMoneyPerson.getName());
        accRes.setMaxMoney(Math.max(accRes.getMaxMoney(),acc.maxMoneyPerson.getMoney()));
        accRes.setMinMoney(Math.min(accRes.getMinMoney(),acc.minMoneyPerson.getMoney()));
        return accRes;
    }

    @Override
    public MyAcc merge(MyAcc acc1, MyAcc acc2) {
        MyAcc acc=new MyAcc();
        acc.avgMoney=(acc1.avgMoney*acc1.count+acc2.avgMoney*acc2.count)/(acc1.count+acc2.count);
        acc.count=acc1.count+acc2.count;
        acc.maxMoneyPerson=acc1.maxMoneyPerson.getMoney()>acc2.maxMoneyPerson.getMoney()?acc1.maxMoneyPerson:acc2.maxMoneyPerson;
        acc.minMoneyPerson=acc1.minMoneyPerson.getMoney()<acc2.minMoneyPerson.getMoney()?acc1.minMoneyPerson:acc2.minMoneyPerson;
        return acc;
    }

    /**
     * 自定义中间状态
     */
    public static class MyAcc{
        //月收入平均值
        public double avgMoney;
        //人数
        public int count;
        //月收入最高的人
        public Person maxMoneyPerson;
        //月收入最低的人
        public Person minMoneyPerson;


    }
}
